//
// Created by guo on 24-8-29.
//

#ifndef TCPSERVER_H
#define TCPSERVER_H

#include <string>
#include <iostream>
#include <sstream>
#include <thread>
#include <unordered_map>

#include <boost/asio.hpp>
#include <boost/make_shared.hpp>
#include "lockfree_queue.h"
#include "struct.h"

using namespace boost::asio;

#define BUFFER_SIZE 1024

class Session {
public:
    explicit Session(io_service &service,boost::shared_ptr<LockFreeQueue<QueueDataPtr>>& ptr_raw_data_queue) :
    _socket(service),
    _ptr_raw_data_queue(ptr_raw_data_queue) {
        _data.resize(BUFFER_SIZE);
    }
    Session(Session&& other) noexcept :_socket(std::move(other._socket)) {
        _data.resize(BUFFER_SIZE);
        memcpy(_data.data(), other._data.data(),BUFFER_SIZE);
        _ptr_raw_data_queue = other._ptr_raw_data_queue;
    }

    ~Session(){
    }

    ip::tcp::socket &socket(){
        return _socket;
    }

    void send(const std::string& data){
        _socket.async_send(boost::asio::buffer(data),
                           [](boost::system::error_code ec, size_t size) {
                               if (ec) {
                                   std::cout << ec.message() << std::endl;
                               } else {
                                   //std::cout << "send size: " << size << std::endl;
                               }
                           });
    }

    void start() {
        async_read();
    }

    std::string session_name(){
        auto endpoint = _socket.remote_endpoint();
        return (std::ostringstream()
                << endpoint.address().to_string()
                << ":"
                << endpoint.port()).str();
    }

    bool is_open(){
        return _socket.is_open();
    }

protected:
    void async_read() {
        memset(_data.data(),0,BUFFER_SIZE);
        _socket.async_read_some(boost::asio::buffer(_data),
                                std::bind(&Session::handle_read,
                                          this, std::placeholders::_1, std::placeholders::_2));
    }

    void handle_read(const boost::system::error_code &error, size_t bytes_transferred) {
        if (!error) {
            // 处理接收到的数据
            //std::cout << session_name() << std::endl;
            QueueDataPtr queue_data = boost::shared_ptr<QueueData>(new QueueData);
            queue_data->session_name = session_name();
            queue_data->data.resize(_data.size());
            memcpy((void *) queue_data->data.data(), _data.data(), _data.size());
            _ptr_raw_data_queue->enqueue(queue_data);
            async_read(); // 继续接收数据
        } else {
            std::cout << error.message() << std::endl;
            if(_socket.is_open()){
                //_socket.shutdown(ip::tcp::socket::shutdown_receive);
                _socket.close();
            }
//            delete this;
        }
    }

private:
    ip::tcp::socket _socket;
    std::vector<char> _data;
    boost::shared_ptr<LockFreeQueue<QueueDataPtr>> _ptr_raw_data_queue;
};

class TCPServer {
public:
    struct Msg{
        std::string session_name;
        std::string data;
    };

    TCPServer(const std::string &ip, uint16_t port) :
            _work(_service),
            _acceptor(_service, ip::tcp::endpoint(ip::address::from_string(ip), port)),
            _raw_data_queue(boost::make_shared<LockFreeQueue<QueueDataPtr>>()){

    }

    void run(int thread_num) {
        std::vector<std::thread> threads;
        threads.reserve(thread_num);
        for (int i = 0; i < thread_num; i++) {
            threads.emplace_back([this]() {
                while (1) {
                    _service.poll_one();
                    std::this_thread::sleep_for(std::chrono::milliseconds(80));
                }
            });
        }

        std::thread _session_control_thread([this](){
            while(1){
                for(auto it = _session_map.begin();it != _session_map.end();){
                    if(it->second->is_open()){
                        it++;
                    }else{
                        it = _session_map.erase(it);
                    }
                }
                std::this_thread::yield();
            }
        });

        start_accept();

        for (auto &t: threads) {
            t.join();
        }
        _session_control_thread.join();
    }

    bool send(const std::string& session_name, const std::string& data){
        auto it = _session_map.find(session_name);
        if(it == _session_map.end()){
            return false;
        }else{
            it->second->send(data);
            return true;
        }
    }

    bool get_data(QueueDataPtr& queue_data){
        bool ret = _raw_data_queue->dequeue(queue_data);
        if(ret){
            return true;
        }else{
            return false;
        }
    }

protected:
    void start_accept() {
        auto session = boost::make_shared<Session>(_service,_raw_data_queue);
        _acceptor.async_accept(session->socket(),
                               std::bind(&TCPServer::handle_accept, this, session, std::placeholders::_1));
    }

    void handle_accept(boost::shared_ptr<Session>& new_session, const boost::system::error_code &error) {
        if (!error && !has_session(new_session->session_name())) {
            std::cout << "session connected !!!" << std::endl;
            std::string session_name = new_session->session_name();
            add_session(new_session);
            _session_map.at(session_name)->start();
            start_accept(); // 继续接受新的连接
        } else {
            std::cout << error.message() << std::endl;
        }
    }
    bool has_session(const std::string& name){
        return (_session_map.find(name) != _session_map.end());
    }

    bool add_session(boost::shared_ptr<Session>& session){
        if(_session_map.find(session->session_name()) != _session_map.end()){
            return false;
        }else{
            _session_map.insert(std::make_pair(session->session_name(), session));
            return true;
        }
    }

private:
    io_service _service;
    boost::asio::io_service::work _work;
    ip::tcp::acceptor _acceptor;
    std::unordered_map<std::string, boost::shared_ptr<Session>> _session_map;
    boost::shared_ptr<LockFreeQueue<QueueDataPtr>> _raw_data_queue;
};


#endif //TCPSERVER_H
